package org.tao.lightningmq.broker.timewheel;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.broker.event.model.TimeWheelEvent;
import org.tao.lightningmq.common.event.EventBus;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @Author lht
 * @date 2025/1/22 - 14:14
 * @description:
 */
public class TimeWheelModelManager {

    private static final Logger LOGGER = LoggerFactory.getLogger(TimeWheelModelManager.class);

    private long executeSeconds = 0;
    private TimeWheelModel secondsTimeWheelModel;
    private TimeWheelModel minutesTimeWheelModel;
    private EventBus eventBus;
    private ExecutorService executorService;

    private final Object secondsLock = new Object();
    private final Object minutesLock = new Object();

    public void init(EventBus eventBus) {
        secondsTimeWheelModel = new TimeWheelModel();
        secondsTimeWheelModel.setUnit(TimeWheelSlotStepUnitEnum.SECOND.getCode());
        secondsTimeWheelModel.setCurrent(0);
        secondsTimeWheelModel.setTimeWheelSlotListModel(buildTimeWheelSlotListModel(60));

        minutesTimeWheelModel = new TimeWheelModel();
        minutesTimeWheelModel.setCurrent(0);
        minutesTimeWheelModel.setTimeWheelSlotListModel(buildTimeWheelSlotListModel(60));
        minutesTimeWheelModel.setUnit(TimeWheelSlotStepUnitEnum.MINUTE.getCode());
        this.eventBus = eventBus;
        this.eventBus.init();
        executorService = Executors.newFixedThreadPool(2);
        LOGGER.info("time-wheel init success");
    }

    private TimeWheelSlotListModel[] buildTimeWheelSlotListModel(int count) {
        TimeWheelSlotListModel[] timeWheelSlotListModels = new TimeWheelSlotListModel[count];
        for (int i = 0; i < count; i++) {
            timeWheelSlotListModels[i] = new TimeWheelSlotListModel();
        }
        return timeWheelSlotListModels;
    }

    public void addDelayMessage(DelayMessageDTO delayMessageDTO) {
        int delay = delayMessageDTO.getDelay();
        if (delay < 0) {
            throw new IllegalArgumentException("delay can not less than 0");
        }
        int min = delay / 60;
        TimeWheelSlotModel timeWheelSlotModel = TimeWheelSlotModel.fromDelayMessageDTO(delayMessageDTO);
        if (min > 0) {
            int nextSlot = minutesTimeWheelModel.countNextSlot(min);
            LOGGER.info("current minutes slot:{},next slot:{}", minutesTimeWheelModel.getCurrent(), nextSlot);
            TimeWheelSlotListModel timeWheelSlotListModel = minutesTimeWheelModel.getTimeWheelSlotListModel()[nextSlot];
            synchronized (minutesLock) {
                timeWheelSlotListModel.getTimeWheelSlotModels().add(timeWheelSlotModel);
            }
        } else {
            int nextSlot = secondsTimeWheelModel.countNextSlot(delay);
            LOGGER.info("current second slot:{},next slot:{}", secondsTimeWheelModel.getCurrent(), nextSlot);
            TimeWheelSlotListModel timeWheelSlotListModel = secondsTimeWheelModel.getTimeWheelSlotListModel()[nextSlot];
            synchronized (secondsLock) {
                timeWheelSlotListModel.getTimeWheelSlotModels().add(timeWheelSlotModel);
            }
        }
    }

    public void doScanTask() {
        new Thread(() -> {
            LOGGER.info("start scan slot task");
            while (true) {
                try {
                    // 每秒执行秒级时间轮任务
                    executorService.execute(this::doSecondsTimeWheelExecute);
                    //过去了一分钟，执行分钟级别的时间轮
                    if (executeSeconds % 60 == 0) {
                        executorService.execute(this::doMinutesTimeWheelExecute);
                    }
                    TimeUnit.SECONDS.sleep(1);
                    executeSeconds++;
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }, "scan-slot-task").start();
    }

    private void doMinutesTimeWheelExecute() {
        int current = minutesTimeWheelModel.getCurrent();
        TimeWheelSlotListModel timeWheelSlotListModel = minutesTimeWheelModel.getTimeWheelSlotListModel()[current];
        List<TimeWheelSlotModel> timeWheelSlotModelList = null;
        // 取出当前slot的数据，随后清空
        synchronized (minutesLock) {
            timeWheelSlotModelList = timeWheelSlotListModel.getTimeWheelSlotModels();
            timeWheelSlotListModel.setTimeWheelSlotModels(new ArrayList<>());
        }
        // 对于延时时间未到的数据，重新计算延时时间，投递到秒级时间轮中
        // 对于延时时间到的数据，收集到集合中并发布事件
        timeWheelSlotModelList = timeWheelSlotModelList.stream().filter(item -> {
            int delay = item.getDelaySeconds();
            int sec = delay % 60;
            // sec != 0 需要投递到秒级的时间轮中
            if (sec != 0) {
                DelayMessageDTO delayMessageDTO = new DelayMessageDTO();
                delayMessageDTO.setDelay(sec);
                delayMessageDTO.setSlotStoreType(SlotStoreTypeEnum.MESSAGE_RETRY_DTO);
                delayMessageDTO.setData(item.getData());
                delayMessageDTO.setNextExecuteTime(item.getNextExecuteTime());
                addDelayMessage(delayMessageDTO);
            }
            return sec == 0;
        }).collect(Collectors.toList());
        // 如果当前slot还有数据，发布事件
        if (!timeWheelSlotModelList.isEmpty()) {
            TimeWheelEvent timeWheelEvent = new TimeWheelEvent();
            timeWheelEvent.setTimeWheelSlotModelList(timeWheelSlotModelList);
            eventBus.publish(timeWheelEvent);
        }
        // 分钟数 + 1
        current = (current + 1) % minutesTimeWheelModel.getTimeWheelSlotListModel().length;
        minutesTimeWheelModel.setCurrent(current);
    }

    private void doSecondsTimeWheelExecute() {
        int current = secondsTimeWheelModel.getCurrent();
        TimeWheelSlotListModel timeWheelSlotListModel = secondsTimeWheelModel.getTimeWheelSlotListModel()[current];
        List<TimeWheelSlotModel> timeWheelSlotModelList = null;
        // 取出当前slot的数据，随后清空
        synchronized (secondsLock) {
            timeWheelSlotModelList = timeWheelSlotListModel.getTimeWheelSlotModels();
            timeWheelSlotListModel.setTimeWheelSlotModels(new ArrayList<>());
        }
        // 如果当前slot有数据，发布事件
        if (timeWheelSlotModelList != null && !timeWheelSlotModelList.isEmpty()) {
            TimeWheelEvent timeWheelEvent = new TimeWheelEvent();
            timeWheelEvent.setTimeWheelSlotModelList(timeWheelSlotModelList);
            eventBus.publish(timeWheelEvent);
        }
        // 秒数 + 1
        current = (current + 1) % secondsTimeWheelModel.getTimeWheelSlotListModel().length;
        secondsTimeWheelModel.setCurrent(current);
    }

}
